package org.example.modelTest;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class MqttTest {
    private final static int QOS_1 = 1;
    private final static String USER_NAME = "zmh";
    private final static int PASSWORLD = 123456;
    private final static int KEEP_ALIVE = 60;
    //客户端唯一ID
    private static final String clientid = "11144";


    private boolean isMqttClientCreated = false;
    private MqttClient mqttClient = null;
    private final String BROKER = "tcp://47.92.164.101:1884";
    private final String PUB_TOPIC = "testTopic";
    private final String SUB_TOPIC = "responseTopic";
    private final int TIMEOUT_MS = 2000;//最多等两秒
    public static void main(String[] args) throws Exception {
        MqttTest mqttTest=new MqttTest();
        String response=mqttTest.sendControlMessage("787986992669","test");
        System.out.println(response);
    }
    public String sendControlMessage(String sn,String controlCode) throws Exception {
        String topicPub="v1/" + sn + "/me/rpc/request";
        String topicSub="v1/" + sn + "/me/telemetry";
        // 创建MqttClient
        if (!isMqttClientCreated) {
            init();
            isMqttClientCreated = true;
        }
        String msg="\"Method\":\"DTU\",\"DATA\":\"CCDD0D570100000000000E0D0A\"";//用controlCode构造code57，目前是测试
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(QOS_1);
        mqttMessage.setPayload(msg.getBytes());
        // 发布控制消息
        try {
            mqttClient.publish(topicPub, mqttMessage);
            System.out.println("MQTT客户端:" + clientid + "向服务器发送消息:{}" + mqttMessage);
        } catch (MqttException e) {
            System.out.println("MQTT消息推送失败");
            e.printStackTrace();
        }

        // 订阅响应主题并等待消息返回
        CountDownLatch latch = new CountDownLatch(1);
        final String[] response = {null};
        mqttClient.subscribe(topicSub, (topic, message) -> {
            // 判断是否为响应主题以及消息是否以0x53开头
            if (topic.equals(topicSub) && message.toString().contains("DTU")&&message.toString().substring(30,32).equals("57")) {
                // 消息符合要求，设置响应消息并countDown
                response[0] = message.toString();
                latch.countDown();
            }
        });

        // 等待超时或者消息返回
//        String response = null;
        if (latch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
            // 如果等待latch计数归零并且收到响应消息，返回响应消息
            mqttClient.disconnect();
            mqttClient.close();
            isMqttClientCreated = false;
            return response[0];
        } else {
            // 如果等待超时，返回null
            mqttClient.disconnect();
            mqttClient.close();
            isMqttClientCreated = false;
            return null;
        }
    }
    private void init() throws Exception {
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        //断开之后自动重联
        connectOptions.setAutomaticReconnect(false);//true改为了flase @pulsar
        //设置会话心跳时间 代理和客户端在不发送消息的情况下可以忍受的最长时间
        connectOptions.setKeepAliveInterval(KEEP_ALIVE);
        //不建立持久会话
        connectOptions.setCleanSession(true);
        //用户名
        connectOptions.setUserName(USER_NAME);
        //密码
        connectOptions.setPassword(String.valueOf(PASSWORLD).toCharArray());

        String clientId = MqttClient.generateClientId();
        try {
            mqttClient = new MqttClient(BROKER, clientid, new MemoryPersistence());
            //MQTT连接
            mqttClient.connect(connectOptions);
            //消息回调
//            client.setCallback(new MqttCallBackHandle(client));
        } catch (MqttException e) {
            System.out.println("MQTT消息异常{}");
        }
//        mqttClient = new MqttClient(BROKER, clientId);
//        mqttClient.connect();
    }

}
